//package lzzsrj.union.message.test;
//
//import com.alibaba.fastjson.JSON;
//import lombok.SneakyThrows;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.boot.ApplicationArguments;
//import org.springframework.boot.ApplicationRunner;
//import org.springframework.dao.DataAccessException;
//import org.springframework.data.redis.core.RedisOperations;
//import org.springframework.data.redis.core.RedisTemplate;
//import org.springframework.data.redis.core.SessionCallback;
//import org.springframework.stereotype.Component;
//
//import java.util.ArrayList;
//import java.util.Collections;
//import java.util.List;
//import java.util.stream.Collectors;
//
//@Component
//public class RedisMQMessageRegister implements ApplicationRunner {
//
//    @Autowired
//    private RedisTemplate<String, Object> redisTemplate;
//
//
//    List<RedisMQListenerMethod> redisMQListenerMethodList = new ArrayList<>();
//
//
//    @Override
//    public void run(ApplicationArguments args) throws Exception {
//
//        this.redisMQListenerMethodList = RedisListenerScanAnnotationProcessor.getRedisMQListenerMethodList();
//
//        for (RedisMQListenerMethod redisMQListenerMethod : redisMQListenerMethodList) {
//            new Thread(new Worker(
//                    redisMQListenerMethod.getQueue(),
//                    redisMQListenerMethod.getInterval(),
//                    redisMQListenerMethod.getQuantity())).start();
//        }
//    }
//
//    private class Worker implements Runnable {
//
//        private String queueName = "";
//        private Long time;
//        private Integer quantity;
//
//        @SneakyThrows
//        @Override
//        public void run() {
//            while (true) {
//                List<String> msg = multiRPopPipeline(queueName, quantity);
//                List<RedisMQMessage> redisMQMessages = new ArrayList<>();
//                for (String s : msg) {
//                    redisMQMessages.add(JSON.parseObject(s, RedisMQMessage.class));
//                }
//                if (redisMQMessages.size() > 0) {
//                    for (RedisMQListenerMethod redisMQListenerMethod : redisMQListenerMethodList) {
//                        if (redisMQListenerMethod.getQueue().equals(queueName)) {
//                            for (RedisMQMessage redisMQMessage : redisMQMessages) {
//                                redisMQListenerMethod.getTargetMethod().invoke(redisMQListenerMethod.getBean(), redisMQMessage);
//                            }
//                        }
//                    }
//                }
//                Thread.sleep(time);
//            }
//        }
//
//        public Worker(String queueName, Long time, Integer quantity) {
//            this.queueName = queueName;
//            this.time = time;
//            this.quantity = quantity;
//        }
//    }
//
//    /**
//     * 一次性pop出指定数量的数据
//     *
//     * @param key  键
//     * @param size 需要取出的元素个数
//     * @return 返回取出的元素集合
//     */
//    public List<String> multiRPopPipeline(String key, int size) {
//        // 获取当前队列里的值
//        int curSize = Math.toIntExact(redisTemplate.opsForList().size(key));
//        if (curSize == 0) {
//            return Collections.emptyList();
//        }
//        // 判断操作次数
//        List<String> collect = redisTemplate.executePipelined(new SessionCallback<String>() {
//            @Override
//            public String execute(RedisOperations redisOperations) throws DataAccessException {
//                final int finalSize = Math.toIntExact(Math.min(curSize, size));
//                for (int i = 0; i < finalSize; i++) {
//                    redisOperations.opsForList().rightPop(key);
//                }
//                return null;
//            }
//        }).stream().map(obj -> (String) obj).collect(Collectors.toList());
//        return collect;
//    }
//}
